package com.study;

import com.study.entity.UserBehavior;
import com.study.utils.JsonHelper;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author sjw
 * @className SoureKafka
 * @date 2020/11/20 8:13
 * @description: TODO
 */
public class SoureKafkaTimer {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.10.200:9092");
        props.put("zookeeper.connect", "192.168.10.200:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest"); //value 反序列化

        DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer010<String>(
                "logEvent", new SimpleStringSchema(), props)).setParallelism(1);
        dataStreamSource.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                return JsonHelper.jsonToObject(s, UserBehavior.class);
            }
        }).filter(new FilterFunction<UserBehavior>() {
            @Override
            public boolean filter(UserBehavior userBehavior) throws Exception {
                return userBehavior.getBehavior().equals("pv");
            }
        }).keyBy("userId").process(new KeyedProcessFunction<Tuple, UserBehavior, Object>() {


            // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
            private ListState<UserBehavior> userBehaviorListState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                // 状态的注册
                ListStateDescriptor<UserBehavior> itemsStateDesc = new ListStateDescriptor<>(
                        "itemState-state",
                        UserBehavior.class);
                userBehaviorListState = getRuntimeContext().getListState(itemsStateDesc);
            }

            @Override
            public void processElement(
                    UserBehavior input,
                    Context context,
                    Collector<Object> collector) throws Exception {

                // 每条数据都保存到状态中
                System.out.println(" processElement "+context.getCurrentKey() +"  " + input.userId + "   " + input.timestamp + 10000);
                userBehaviorListState.add(input);
                // 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据


                //context.timerService().registerEventTimeTimer(System.currentTimeMillis() + 10000);
//                input.itemId=input.timestamp + 10000;
                //context.timerService().registerProcessingTimeTimer(input.timestamp + 10000);
                context.timerService().registerEventTimeTimer(input.timestamp + 10000);
            }

            @Override
            public void onTimer(
                    long timestamp, KeyedProcessFunction<Tuple, UserBehavior, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
                System.out.println("timer");
                List<UserBehavior> userBehaviors=new ArrayList<>();
                for(UserBehavior userBehavior:userBehaviorListState.get()){
                    //System.out.println(ctx.timerService().currentProcessingTime()+"   "+userBehavior.itemId+"   "+(ctx.timerService().currentProcessingTime()-userBehavior.itemId));
                    //System.out.println(ctx.timerService().currentProcessingTime()-userBehavior.itemId);
                    userBehaviors.add(userBehavior);
                }
                if(userBehaviors.size()==0)return;
                System.out.println( "timer-------------"+userBehaviors.size()+"   "+ctx.timerService().currentProcessingTime() +"   "+timestamp);
                userBehaviorListState.clear();
                out.collect(userBehaviors);
            }
        }).print("print-----");

        // singleOutputStreamOperator.print();


        env.execute("Flink add data source kafak ");
    }


}
